package com.atguigu.ad.flume;

import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @author Hefei
 * @description Flume中的时间拦截器
 * @project_name com.atguigu.ad.flume
 * @since 2023/4/5 15:59
 */
public class TimestampInterceptor implements Interceptor {
    private Pattern pattern;

    @Override
    public void initialize() {
        pattern = Pattern.compile(".*t=(\\d{13}).*");
    }

    @Override
    public Event intercept(Event event) {
        // todo 1. 把数据拿到
        String log = new String(event.getBody());

        // todo 2. 处理 去掉引号
        // "2023-01-07T13:07:48.601""GET""/ad/juliang/impression?id=44&t=1673068067304&ip=203.32.48.162&ua=HubSpot%20Connect%202.0%20(http://dev.hubspot.com/)%20-%20BizOpsCompanies-Tq2-BizCoDomainValidationAudit&device_id=&os_type=""200""203.195.136.146"

        String subLog = log.substring(1, log.length() - 1);
        String result = subLog.replaceAll("\"\\u0001\"", "\u0001");
        event.setBody(result.getBytes(StandardCharsets.UTF_8));

        // todo 3. 使用正则的分组捕获功能  拿到t= 后面的 13位数字
        // 为了能在全局使用Pattern对象，将变量提升为全局变量

        Matcher matcher = pattern.matcher(result);



        // todo 4. 放入到event的header中
        if (matcher.matches()) {
            String ts = matcher.group(1);
            event.getHeaders().put("timestamp", ts);
        } else {
            return  null;
        }
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> events) {
        List<Event> resultList = new ArrayList<>();
        for (Event event : events) {
            Event intercept = intercept(event);
            if (intercept != null) {
                resultList.add(intercept);
            }
        }

        return resultList;
    }

    @Override
    public void close() {

    }

    public  static  class  Builder implements  Interceptor.Builder{
        @Override
        public Interceptor build() {
           return new TimestampInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }
}